package com.atguigu.gmall.realtime.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * Author: Felix
 * Date: 2022/4/1
 * Desc: 操作kafka的工具类
 */
public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop202:9092,hadoop203:9092,hadoop204:9092";
    private static final String DEFAULT_TOPIC = "default_topic";

    //获取消费者对象
    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), props);
        return kafkaConsumer;
    }

    /* //获取生产者对象
     public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
         FlinkKafkaProducer<String> flinkKafkaProducer =
             new FlinkKafkaProducer<>(KAFKA_SERVER, topic, new SimpleStringSchema());
         return flinkKafkaProducer;
     }*/
    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");

        FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer<String>(
            DEFAULT_TOPIC,
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
                    return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
                }
            }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return flinkKafkaProducer;
    }

    //获取生产者对象  ----- 自定义序列化过程
    public static <T> FlinkKafkaProducer<T> getKafkaSinkSchema(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");

        FlinkKafkaProducer<T> flinkKafkaProducer = new FlinkKafkaProducer<T>(DEFAULT_TOPIC,
            kafkaSerializationSchema, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return flinkKafkaProducer;
    }

    //获取flinkSQL连接kafka的连接属性配置
    public static String getKafkaDDL(String topic, String groupId) {
        String ddl =
            "  'connector' = 'kafka'," +
            "  'topic' = '" + topic + "'," +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "'," +
            "  'properties.group.id' = '" + groupId + "'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'";
        return ddl;
    }
}
